package com.bw.fun;

import com.bw.Test.DwsTradeOrgOrderDayBean;
import com.bw.bean.OrderBoundWide;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;

public class AsyncIOFuncykwide extends RichAsyncFunction<DwsTradeOrgOrderDayBean, DwsTradeOrgOrderDayBean> {
    Connection conn = null;
    PreparedStatement ps1 = null;
    PreparedStatement ps2 = null;


    // 从Mysql查
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/tms01?characterEncoding=utf8", "root", "123456");
        ps1 = conn.prepareStatement("select org_name,id from base_organ where region_id = ?");
        ps2 = conn.prepareStatement("select name from base_region_info where id = ?");
    }

    @Override
    public void close() throws Exception {
        ps1.close();
        ps2.close();
        conn.close();
    }

    @Override
    public void asyncInvoke(DwsTradeOrgOrderDayBean orderInfoWide, ResultFuture<DwsTradeOrgOrderDayBean> resultFuture) throws Exception {

        ps1.setLong(1,Long.parseLong(orderInfoWide.getSenderDistrictId()));
        ResultSet rs1 = ps1.executeQuery();
        if(rs1.next()){
            String id = rs1.getString("id");
            String org_name = rs1.getString("org_name");

            orderInfoWide.setOrgName(org_name);
            orderInfoWide.setOrgId(id);

            ps2.setLong(1,Long.parseLong(orderInfoWide.getCityId()));
            ResultSet rs2 = ps2.executeQuery();
            if(rs2.next()){
                orderInfoWide.setCityName(rs2.getString("name"));
            }

            resultFuture.complete(Collections.singleton(orderInfoWide));
        }
    }
}
